/////////////////////////////////////////////////////////////////////////////
// Original code from libhdfs3. Copyright (c) 2013 - 2014, Pivotal Inc.
// All rights reserved. Author: Zhanwei Wang
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#pragma once

#include <vector>
#include <stdint.h>
#include <cstdlib>

#include <kmhdfs/network/socket.h>

namespace Hdfs {
    namespace Internal {
        /**
         * A warper of Socket, read big endian int and varint from socket.
         */
        class BufferedSocketReader {
        public:
            virtual ~BufferedSocketReader() {
            }

            /**
             * Read data from socket, if there is data buffered, read from buffer first.
             * If there is nothing can be read, the caller will be blocked.
             * @param b The buffer used to receive data.
             * @param s The size of bytes to be read.
             * @return The size of data already read.
             * @throw HdfsNetworkException
             * @throw HdfsEndOfStream
             */
            virtual int32_t read(char *b, int32_t s) = 0;

            /**
             * Read data form socket, if there is data buffered, read from buffer first.
             * If there is not enough data can be read, the caller will be blocked.
             * @param b The buffer used to receive data.
             * @param s The size of bytes to read.
             * @param timeout The timeout interval of this read operation, negative means infinite.
             * @throw HdfsNetworkException
             * @throw HdfsEndOfStream
             * @throw HdfsTimeout
             */
            virtual void readFully(char *b, int32_t s, int timeout) = 0;

            /**
             * Read a 32 bit big endian integer from socket.
             * If there is not enough data can be read, the caller will be blocked.
             * @param timeout The timeout interval of this read operation, negative means infinite.
             * @return A 32 bit integer.
             * @throw HdfsNetworkException
             * @throw HdfsEndOfStream
             * @throw HdfsTimeout
             */
            virtual int32_t readBigEndianInt32(int timeout) = 0;

            /**
             * Read a variable length encoding 32bit integer from socket.
             * If there is not enough data can be read, the caller will be blocked.
             * @param timeout The timeout interval of this read operation, negative means infinite.
             * @return A 32 bit integer.
             * @throw HdfsNetworkException
             * @throw HdfsEndOfStream
             * @throw HdfsTimeout
             */
            virtual int32_t readVarint32(int timeout) = 0;

            /**
             * Test if the socket can be read without blocking.
             * @param timeout Time timeout interval of this operation, negative means infinite.
             * @return Return true if the socket can be read without blocking, false on timeout.
             * @throw HdfsNetworkException
             * @throw HdfsTimeout
             */
            virtual bool poll(int timeout) = 0;
        };

        /**
         * An implement of BufferedSocketReader.
         */
        class BufferedSocketReaderImpl : public BufferedSocketReader {
        public:
            BufferedSocketReaderImpl(Socket &s);

            BufferedSocketReaderImpl(Socket &s, size_t bufferSize);

            int32_t read(char *b, int32_t s);

            void readFully(char *b, int32_t s, int timeout);

            int32_t readBigEndianInt32(int timeout);

            int32_t readVarint32(int timeout);

            bool poll(int timeout);

        private:
            int32_t readVarint32(int timeout, int32_t step);

            //for test
            BufferedSocketReaderImpl(Socket &s, const std::vector<char> &buffer) : cursor(0), size(buffer.size()),
                sock(s), buffer(buffer) {
            }

        private:
            int32_t cursor;
            int32_t size;
            Socket &sock;
            std::vector<char> buffer;
        };
    }
}
